package com.jzsec.proxy.jzsecproxy.consumer;


import com.jzsec.proxy.jzsecproxy.limit.RedisLimit;
import com.jzsec.proxy.jzsecproxy.services.LeancloudService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import redis.clients.jedis.exceptions.JedisConnectionException;

import javax.annotation.PostConstruct;

/**
 * 消费list里面的消息，左进右出
 * 测试用 redis 命令
 * lpush leancloud-req-list "{\"reqHeader\":{\"X-LC-Id\":\"pYRbJzUEENH1FeOhwBzwBM4p-gzGzoHsz\",\"X-LC-Key\":\"3w9fEyL8I7neyP696MKRPw1j,master\",\"Content-Type\":\"application/json\"},\"reqBody\":{\"from_peer\":\"sys_secretary_001\",\"message\":{\"_lctype\":-1,\"_lctext\":\"This is a java msg for test24\",\"_lcattrs\":{\"content\":\"This is a java msg for test24\"}},\"conv_id\":\"569629a260b2a781fa1c768c\",\"transient\":false,\"to_peers\":[\"imaster_67987\"]},\"url\":\"https://leancloud.cn/1.1/rtm/messages\",\"method\":\"POST\"}"
 */
@Component
@Slf4j
public class RedisMsgConsumer {

	private String leancloudRequestQueue = "leancloud-req-list";
	@Autowired
	private LeancloudService leancloudService;
	@Autowired
	private RedisTemplate redisTemplate;
	@Autowired
	private RedisLimit redisLimit;

	@Autowired
	private Environment env;
	@PostConstruct
	private void init() {
		leancloudRequestQueue = env.getProperty("consumer.leancloud-request-queue", leancloudRequestQueue);
	}

	/**
	 * @question msg 堆积太多的时候，线程不够处理，会造成线程池丢弃发送任务，需要处理
	 * @answer 线程池使用 CallerRunsPolicy 饱和策略，线程池满时，使用main线程执行任务，顺便阻塞主线程，不能去取新消息
	 */
	public void start() {
		log.info("========== RedisMsgConsumer started ==============");
		while (true) {
			Object obj = null;
			try {
				obj = redisTemplate.opsForList().rightPop(leancloudRequestQueue);
			} catch (Exception e) {
				log.error("redis execute error", e);
			}

			if (null == obj) {
				try {
					Thread.sleep(1000);
					// log.info(leancloudReqQueuq + " is null, sleep 1s.");
				} catch (InterruptedException e) {
					log.error("", e);
				}
				continue;
			}

			try {
				while (!redisLimit.acquire()) {
					log.info("sleep 10s to wait redis token");
					try {
						Thread.sleep(10000);
					} catch (Exception e) {
						log.error("", e);
					}
				}
			} catch (Exception e) {
				log.error("", e);
			}

			String msg = "";
			try {
				msg = String.valueOf(obj);
				leancloudService.sendMessageByOkHttp(msg);
			} catch (Exception e) {
				log.error(msg, e);
			}
		}
	}
}
